package com.bujunjie.study.realtime.dws.app;

import cn.hutool.core.date.DatePattern;
import cn.hutool.core.date.DateUtil;
import cn.hutool.core.util.ObjectUtil;
import cn.hutool.core.util.StrUtil;
import com.alibaba.fastjson.JSONObject;
import com.bujunjie.study.realtime.common.base.BaseApp;
import com.bujunjie.study.realtime.common.constant.FlinkConstant;
import com.bujunjie.study.realtime.common.function.Object2JsonStringFunction;
import com.bujunjie.study.realtime.common.model.sink.PageInfoModel;
import com.bujunjie.study.realtime.common.model.vo.CommonInfoVO;
import com.bujunjie.study.realtime.common.model.vo.PageInfoVO;
import com.bujunjie.study.realtime.common.util.FlinkSinkUtil;
import com.bujunjie.study.realtime.dws.app.model.sink.TrafficPageViewModel;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple4;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.util.Date;

/**
 * <p>
 * 按照版本、地区、渠道、新老访客对pv、uv、sv、dur进行聚合统计
 * </p>
 *
 * @author bu.junjie
 * @version 1.0.0
 * @createTime 2025/9/22 14:45
 */
public class DwsTrafficVcChArIsNewPageViewWindow extends BaseApp {
    public static void main(String[] args) throws Exception {
        new DwsTrafficVcChArIsNewPageViewWindow().start(
                10022,
                4,
                "dws_traffic_vc_ch_ar_is_new_page_view_window",
                FlinkConstant.TOPIC_DWD_TRAFFIC_PAGE
        );
    }

    @Override
    public void handle(StreamExecutionEnvironment env, DataStreamSource<String> kafkaStrDS) {
        KeyedStream<PageInfoModel, String> midKeyDS = kafkaStrDS.map(x -> JSONObject.parseObject(x, PageInfoModel.class))
                .keyBy(x -> x.getCommon().getMid());

        SingleOutputStreamOperator<TrafficPageViewModel> mapDS = midKeyDS.map(new RichMapFunction<PageInfoModel, TrafficPageViewModel>() {

            private ValueState<Long> lastVisitDateState;

            @Override
            public void open(Configuration parameters) throws Exception {
                ValueStateDescriptor<Long> lastVisitDateStateDesc = new ValueStateDescriptor<>("lastVisitDateState", Long.class);
                lastVisitDateState = getRuntimeContext().getState(lastVisitDateStateDesc);
            }

            @Override
            public TrafficPageViewModel map(PageInfoModel value) throws Exception {
                PageInfoVO page = value.getPage();
                CommonInfoVO common = value.getCommon();
                Long lastVisitDate = lastVisitDateState.value();
                Long ts = value.getTs();
                Long uvCt = 0L;
                if (ObjectUtil.isNull(lastVisitDate) || !lastVisitDate.equals(ts)) {
                    uvCt = 1L;
                    lastVisitDateState.update(ts);
                }
                String lastPageId = page.getLastPageId();
                Long svCt = StrUtil.isNotBlank(lastPageId) ? 1L : 0L;

                TrafficPageViewModel model = TrafficPageViewModel.builder()
                        .stt(StrUtil.EMPTY)
                        .edt(StrUtil.EMPTY)
                        .cur_date(StrUtil.EMPTY)
                        .vc(common.getVc())
                        .ch(common.getCh())
                        .ar(common.getAr())
                        .isNew(common.getIsNew())
                        .uvCt(uvCt)
                        .svCt(svCt)
                        .pvCt(1L)
                        .durSum(page.getDuringTime())
                        .ts(ts)
                        .build();
                return model;
            }
        });

        // 4.指定Watermark以及提取事件时间字段
        SingleOutputStreamOperator<TrafficPageViewModel> withWatermarkDS = mapDS.assignTimestampsAndWatermarks(
                //WatermarkStrategy.<TrafficPageViewBean>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                WatermarkStrategy
                        .<TrafficPageViewModel>forMonotonousTimestamps()
                        .withTimestampAssigner(
                                new SerializableTimestampAssigner<TrafficPageViewModel>() {
                                    @Override
                                    public long extractTimestamp(TrafficPageViewModel bean, long recordTimestamp) {
                                        return bean.getTs();
                                    }
                                }
                        )
        );

        KeyedStream<TrafficPageViewModel, Tuple4<String, String, String, String>> dimKeyedDS = withWatermarkDS.keyBy(
                new KeySelector<TrafficPageViewModel, Tuple4<String, String, String, String>>() {
                    @Override
                    public Tuple4<String, String, String, String> getKey(TrafficPageViewModel bean) throws Exception {
                        return Tuple4.of(bean.getVc(),
                                bean.getCh(),
                                bean.getAr(),
                                bean.getIsNew());
                    }
                }
        );

        WindowedStream<TrafficPageViewModel, Tuple4<String, String, String, String>, TimeWindow> windowDS
                = dimKeyedDS.window(TumblingEventTimeWindows.of(org.apache.flink.streaming.api.windowing.time.Time.seconds(10)));

        // 7.聚合计算
        SingleOutputStreamOperator<TrafficPageViewModel> reduceDS = windowDS.reduce(
                new ReduceFunction<TrafficPageViewModel>() {
                    @Override
                    public TrafficPageViewModel reduce(TrafficPageViewModel value1, TrafficPageViewModel value2) throws Exception {
                        value1.setPvCt(value1.getPvCt() + value2.getPvCt());
                        value1.setUvCt(value1.getUvCt() + value2.getUvCt());
                        value1.setSvCt(value1.getSvCt() + value2.getSvCt());
                        value1.setDurSum(value1.getDurSum() + value2.getDurSum());
                        return value1;
                    }
                },
                new WindowFunction<TrafficPageViewModel, TrafficPageViewModel, Tuple4<String, String, String, String>, TimeWindow>() {
                    @Override
                    public void apply(Tuple4<String, String, String, String> stringStringStringStringTuple4, TimeWindow window, Iterable<TrafficPageViewModel> input, Collector<TrafficPageViewModel> out) throws Exception {
                        TrafficPageViewModel pageViewBean = input.iterator().next();
                        String stt = DateUtil.format(new Date(window.getStart()), DatePattern.NORM_DATETIME_PATTERN);
                        String edt = DateUtil.format(new Date(window.getEnd()), DatePattern.NORM_DATETIME_PATTERN);
                        String curDate = DateUtil.format(new Date(window.getEnd()), DatePattern.NORM_DATE_PATTERN);
                        pageViewBean.setStt(stt);
                        pageViewBean.setEdt(edt);
                        pageViewBean.setCur_date(curDate);
                        out.collect(pageViewBean);
                    }
                }
        );

        reduceDS
                //在向Doris写数据前，将流中统计的实体类对象转换为json格式字符串
                .map(new Object2JsonStringFunction<TrafficPageViewModel>())
                .sinkTo(FlinkSinkUtil.getDorisSink("dws_traffic_vc_ch_ar_is_new_page_view_window"));
    }
}
